package com.thiswhyme.republish.rabbitmq.core;

import com.thiswhyme.republish.common.RedisKey;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.core.io.ClassPathResource;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.scripting.support.ResourceScriptSource;
import org.springframework.util.CollectionUtils;

import java.util.ArrayList;
import java.util.Collections;
import java.util.Date;
import java.util.List;

/**
 * @description: 重发操作redis服务
 * @author: thiswhyme
 * @date: 2021/3/22
 */
public class RepublishRedisOperation {

    public static final Logger log = LoggerFactory.getLogger(RepublishRunner.class);

    @Value("${spring.application.name}")
    private String applicationName;

    private RedisTemplate redisTemplate;

    public RepublishRedisOperation(RedisTemplate redisTemplate) {
        this.redisTemplate = redisTemplate;
    }

    /**
     * 添加指定uniqueId到待ack集合
     * @param uniqueId message唯一标识
     * @param runTimeMillis 执行时间戳
     */
    public void addToAckZSet(Long uniqueId, long runTimeMillis) {
        redisTemplate.opsForZSet().add(RedisKey.genKey(applicationName, RedisKey.TO_ACK_ZEST), uniqueId, runTimeMillis);
        if(log.isDebugEnabled()) {
            log.debug("将消息【{}】加入待ack的zset，执行时间：{}", uniqueId, new Date(runTimeMillis));
        }
    }

    /**
     * 添加指定uniqueId到待重发集合
     * @param uniqueId message唯一标识
     * @param runTimeMillis 执行时间戳
     */
    public void addToRepublishZSet(Long uniqueId, long runTimeMillis) {
        redisTemplate.opsForZSet().add(RedisKey.genKey(applicationName, RedisKey.TO_REPUBLISH_ZEST), uniqueId, runTimeMillis);
        if(log.isDebugEnabled()) {
            log.debug("将消息【{}】加入待重发的zset，执行时间：{}", uniqueId, new Date(runTimeMillis));
        }
    }

    /**
     * 从待ack和待republish集合中删除指定uniqueId的成员
     * @param uniqueId message唯一标识
     */
    public void removeFromAllZSet(Long uniqueId) {
        List<String> keys = new ArrayList<>(2);
        keys.add(RedisKey.genKey(applicationName, RedisKey.TO_ACK_ZEST));
        keys.add(RedisKey.genKey(applicationName, RedisKey.TO_REPUBLISH_ZEST));
        DefaultRedisScript redisScript = new DefaultRedisScript<>();
        redisScript.setScriptSource(new ResourceScriptSource(new ClassPathResource("lua/removeMessage.lua")));
        redisTemplate.execute(redisScript, keys, uniqueId);
        if(log.isDebugEnabled()) {
            log.debug("将消息【{}】从待ack和待republish的zset移除", uniqueId);
        }
    }

    /**
     * 将指定消息从待ack的zset移动到待republish的zset
     * @param uniqueId
     * @param runTimeMillis
     */
    public void moveToRepublishSet(Long uniqueId, long runTimeMillis) {
        List<String> keys = new ArrayList<>(2);
        keys.add(RedisKey.genKey(applicationName, RedisKey.TO_ACK_ZEST));
        keys.add(RedisKey.genKey(applicationName, RedisKey.TO_REPUBLISH_ZEST));
        DefaultRedisScript redisScript = new DefaultRedisScript<>();
        redisScript.setScriptSource(new ResourceScriptSource(new ClassPathResource("lua/moveToRepublishSetIfAckFail.lua")));
        redisTemplate.execute(redisScript, keys, uniqueId, runTimeMillis);
        if(log.isDebugEnabled()) {
            log.debug("将消息【{}】从待ack的zset移动到待republish的zset，执行时间：{}", uniqueId, new Date(runTimeMillis));
        }
    }

    /**
     * 将当前时间之前未收到ack的消息从待ack的zset移动至待republish的zset
     * @return
     */
    public long moveToRepublishSet() {
        List<String> keys = new ArrayList<>(2);
        keys.add(RedisKey.genKey(applicationName, RedisKey.TO_ACK_ZEST));
        keys.add(RedisKey.genKey(applicationName, RedisKey.TO_REPUBLISH_ZEST));
        DefaultRedisScript<List> redisScript = new DefaultRedisScript<>();
        redisScript.setResultType(List.class);
        redisScript.setScriptSource(new ResourceScriptSource(new ClassPathResource("lua/moveToRepublishSetIfTimeout.lua")));
        List<Long> values = (List<Long>) redisTemplate.execute(redisScript, redisTemplate.getValueSerializer(), redisTemplate.getValueSerializer(), keys, System.currentTimeMillis());
        values = CollectionUtils.isEmpty(values) ? Collections.EMPTY_LIST : values;
        long count = CollectionUtils.isEmpty(values) ? 0 : values.get(0);
        if(count > 0 && log.isDebugEnabled()) {
            log.debug("将{}之前未ack的消息从待ack的zset移动到待republish的zset，数量：{}", new Date(System.currentTimeMillis()), count);
        }
        return count;
    }

    /**
     * 从待重发zset中取出并删除指定时间戳之前的消息
     * @return
     */
    public List<Long> getAndRemoveFromRepublishSet() {
        DefaultRedisScript<List> redisScript = new DefaultRedisScript<>();
        redisScript.setResultType(List.class);
        redisScript.setScriptSource(new ResourceScriptSource(new ClassPathResource("lua/removeAndRtMessage.lua")));
        List<Long> values = (List<Long>) redisTemplate.execute(redisScript, redisTemplate.getValueSerializer(),
                redisTemplate.getValueSerializer(),
                Collections.singletonList(RedisKey.genKey(applicationName, RedisKey.TO_REPUBLISH_ZEST)),
                System.currentTimeMillis());
        values = CollectionUtils.isEmpty(values) ? Collections.EMPTY_LIST : values;
        if(values.size() > 0 && log.isDebugEnabled()) {
            log.debug("将{}之前的消息重发，数量：{}", new Date(System.currentTimeMillis()), values.size());
        }
        return values;
    }
}
